
#include <protocol.hpp>
#include <fs.hpp>

#include <boost/asio.hpp>

#define DATA_FILENAME "../../test/test_suite_xbase/wb"

std::vector<std::string> m_line_vector;
std::vector< std::pair<std::string, std::string> > m_list;

int data_bytes = 10;

void load_data(const std::string &data_file) 
{
	const std::string file_name = DATA_FILENAME;
	fs::file_t file = fs::open(file_name);
	if(file == fs::BAD_FILE)
	{
		std::cout << "open file " << file_name << " failed" << std::endl;
		return;
	}

	const size_t size = 1024 * 1024;
	char buffer[size];
	fs::ssize_t n = 0;
	size_t file_size = 0;
	do {
		n = fs::readn(file, buffer, size);
		if(n < 0)
		{
			std::cout << "read file " << file_name << " failed" << std::endl;
			return;
		}
		file_size += n;

		const char *p1 = buffer;
		for(int i=0; i<n; ++i)
		{
			if(buffer[i] == '\n')
			{
				const char *p2 = buffer + i; // 指向'\n'
				m_line_vector.push_back(std::string(p1, p2));
				p1 = p2 + 1; // 指向'\n'的下一个字符
			}
		}
	} while(static_cast<std::size_t>(n) == size);
	fs::close(file);
	cout << endl << "    " << m_line_vector.size() << " lines loaded." << endl;

	// parse row
	typedef std::vector<std::string>::iterator it_type;
	const it_type iter_end = m_line_vector.end();
	for(it_type iter = m_line_vector.begin(); iter != iter_end; ++iter)
	{
		const std::string::size_type pos = iter->find(' ');
		const std::string row = iter->substr(0, pos);
		std::string value = iter->substr(pos + 1);
		while(value.size() < data_bytes)
		{
			value.append(buffer, std::min(file_size,
						      data_bytes - value.size()));
		}
		m_list.push_back(std::make_pair(row,
						value));
	}
}

std::string tablet_name;

std::size_t fill_request(char *p, std::size_t start, std::size_t end)
{
	boost::asio::mutable_buffer data(p, 1024 * 1024 * 16);
	ByteBuffer bb(writable_bb(data));
	BinOArchive out(bb);
	for(std::size_t i = start; i < end; ++i)
	{
		Header h(WRITE_REQ);
		h.sequence_number = i;

		WriteReq req;
		req.tablet_name = tablet_name;
		Record rec(Key(m_list[i].first, "contents:", 0),
			   Value(m_list[i].second));
		req.records.push_back(rec);

		h.data_length = req.bytes();
		out & h;
		out & req;
	}

	return bb.position();
}

std::string ts_ip;
std::string ts_port;

void send_request(boost::asio::ip::tcp::socket &s,
		  char *p, size_t bytes, size_t n)
{

	// send
	boost::asio::write(s, boost::asio::buffer(p, bytes));
	
	for(std::size_t i = 0;i < n; ++i)
	{
		boost::system::error_code ec;
		boost::asio::read(s,
				  boost::asio::buffer(p, Header::LENGTH),
				  boost::asio::transfer_at_least(Header::LENGTH),
				  ec);
		if(ec)
		{
			cout << "1 " << ec.message() << endl;
		}

		header_wrapper *hr = wrapper_cast<header_wrapper*>(p);
		//cout << hr->m_sequence_number.get() << endl;
		boost::asio::read(s,
				  boost::asio::buffer(p, hr->m_data_length.get()),
				  boost::asio::transfer_at_least(hr->m_data_length.get()),
				  ec);
		if(ec)
		{
			cout << "2 " << ec.message() << endl;
		}
	}
}

int main(int argc, char **argv)
{
	if(argc != 6)
	{
		cout << argv[0] << " <ts_ip> <ts_port> <tablet_name> <n> <bytes>" << std::endl;
		return -1;
	}

	ts_ip = argv[1];
	ts_port = argv[2];
	tablet_name = argv[3];
	int n = atoi(argv[4]);
	data_bytes = atoi(argv[5]);
	if(data_bytes <= 0)
		return -1;

	load_data(std::string(DATA_FILENAME));
	char *storage = new char[1024 * 1024 * 16];

	std::size_t pos = 0;
	std::size_t end = m_list.size();
	std::size_t n_per = n;

	boost::asio::io_service io_service;
	boost::asio::ip::tcp::resolver resolver(io_service);
	boost::asio::ip::tcp::resolver::query query(ts_ip, ts_port);
	boost::asio::ip::tcp::resolver::iterator iter = resolver.resolve(query);

	boost::asio::ip::tcp::socket s(io_service);
	s.connect(*iter);

	
	const uint64_t start_time = cur_micro_seconds();

	std::size_t total_bytes = 0;
	std::size_t request_counter = 0;
	while(pos < end)
	{
		std::size_t n = std::min(n_per, end - pos);
		std::size_t bytes = fill_request(storage, pos, pos + n);
		send_request(s, storage, bytes, n);
		pos += n;
		total_bytes += bytes;
		
		request_counter += n;

		if(request_counter >= 10000)
		{
			request_counter = 0;

			// print
			const uint64_t now_time = cur_micro_seconds();
			const uint64_t used_time = now_time - start_time;
			const int used_seconds = used_time / (1000 * 1000);
			const int used_us = used_time % (1000 * 1000);
			char buf[256];
			snprintf(buf, 256, "%10d.%06d", used_seconds, used_us);
			double speed_bytes = ((double)total_bytes) * 1000 * 1000 / used_time;
			double speed_request = ((double)pos) * 1000 * 1000 / used_time;
			std::cout << buf << "s : "
				  << pos << " requests with "
				  << total_bytes << " bytes, average "
				  << speed_request << " requestes/s and "
				  << speed_bytes << " bytes/s" << std::endl;
		}
	}

	// close socket
	s.close();
	

	// print
	const uint64_t now_time = cur_micro_seconds();
	const uint64_t used_time = now_time - start_time;
	const int used_seconds = used_time / (1000 * 1000);
	const int used_us = used_time % (1000 * 1000);
	char buf[256];
	snprintf(buf, 256, "%10d.%06d", used_seconds, used_us);
	double speed_bytes = ((double)total_bytes) * 1000 * 1000 / used_time;
	double speed_request = ((double)pos) * 1000 * 1000 / used_time;
	std::cout << buf << "s : "
		  << pos << " requests with "
		  << total_bytes << " bytes, average "
		  << speed_request << " requestes/s and "
		  << speed_bytes << " bytes/s" << std::endl;
			
	return 0;
}
